package com.lemon.core.amqp.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ配置
 *
 * @author pan
 * @since 2021-01-10
 */
@Slf4j
@Configuration
@EnableConfigurationProperties({ RabbitProperties.class })
public class RabbitConfiguration {

    @Bean
    @ConditionalOnProperty(prefix = RabbitProperties.PREFIX, name = "instanceId", matchIfMissing = true)
    public ConnectionFactory getConnectionFactory(RabbitProperties rabbitProperties) {
        com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory =
                new com.rabbitmq.client.ConnectionFactory();
        rabbitConnectionFactory.setHost(rabbitProperties.getHost());
        rabbitConnectionFactory.setPort(rabbitProperties.getPort());
        rabbitConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());

        AliyunCredentialsProvider credentialsProvider = new AliyunCredentialsProvider(
                rabbitProperties.getUsername(), rabbitProperties.getPassword(), rabbitProperties.getInstanceId());
        rabbitConnectionFactory.setCredentialsProvider(credentialsProvider);
        rabbitConnectionFactory.setAutomaticRecoveryEnabled(true);
        rabbitConnectionFactory.setNetworkRecoveryInterval(5000);
        rabbitConnectionFactory.setRequestedChannelMax(4095);
        ConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
        ((CachingConnectionFactory) connectionFactory).setPublisherConfirmType(rabbitProperties.getPublisherConfirmType());
        ((CachingConnectionFactory) connectionFactory).setPublisherReturns(rabbitProperties.isPublisherReturns());
        return connectionFactory;
    }

    /**
     * 描述:队列名,定义rabbit template用于数据的接收和发送;
     * 备注: 配置exchange和queue
     *
     * @author lilun
     * @date 2019-09-19 15:12:30
     */
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // 使用jackson 消息转换器
        template.setMessageConverter(producerJackson2MessageConverter());
        template.setEncoding("UTF-8");
        // 消息发送失败返回到队列中，yml需要配置 publisher-returns: true
        template.setMandatory(true);
        // 设置
        template.setReplyTimeout(30 * 60 * 1000L);
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("消息主体 message : " + message);
            log.info("消息主体 message : " + replyCode);
            log.info("描述：" + replyText);
            log.info("消息使用的交换器 exchange : " + exchange);
            log.info("消息使用的路由键 routing : " + routingKey);
        });
        // 消息确认，yml需要配置 publisher-confirms: true
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("消息唯一标识：" + correlationData);
                log.info("确认结果：" + ack);
                log.info("失败原因：" + cause);
            } else {
                log.info("消息发送到exchange失败,原因: " + cause);
            }
        });
        return template;
    }

    @Bean
    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

}
